package com.jeffse.light.stream.core.service;

import com.jeffse.light.stream.core.data.context.TimerContext;
import com.jeffse.light.stream.core.pojo.EventTimer;
import com.jeffse.light.stream.core.pojo.WaterMark;
import com.jeffse.light.stream.core.utils.ScheduleUtils;
import lombok.SneakyThrows;

import java.util.HashMap;
import java.util.Map;

public class SimpleTimerService implements TimerService {
    private Map<String, EventTimer> eventTimerMap = new HashMap<>();
    private TimerContext timerContext;
    private Map<String, WaterMark> waterMarkMap = new HashMap<>();

    public SimpleTimerService(TimerContext timerContext){
        this.timerContext = timerContext;
    }

    @Override
    public long currentProcessingTime() {
        return System.currentTimeMillis();
    }

    @Override
    public long currentWatermark() {
        return 0;
    }

    @Override
    public void registerProcessingTimeTimer(long time) {
        Object currentKey = timerContext.getCurrentKey();
        Runnable func = new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                synchronized (timerContext){
                    timerContext.setCurrentKey(currentKey);
                    timerContext.onTimer(currentProcessingTime(), timerContext, timerContext.getProcessContext());
                    deleteProcessingTimeTimer(time);
                }
            }
        };

        Long delay = time - System.currentTimeMillis();
        ScheduleUtils.register(func, delay, "" + time + "_" + timerContext.getCurrentKey() + "_" + timerContext.getUUID());
    }

    @Override
    public void registerEventTimeTimer(long time) {
        eventTimerMap.put(time + "_" + timerContext.getCurrentKey(), new EventTimer(time, "event", timerContext));
    }

    @Override
    public void deleteProcessingTimeTimer(long time) {
        ScheduleUtils.remove(time + "_" + timerContext.getCurrentKey() + "_" + timerContext.getUUID());
    }

    @Override
    public void deleteEventTimeTimer(long time) {
        eventTimerMap.remove(time + "_" + timerContext.getCurrentKey());
    }

    @Override
    public void triggerEvent() throws Exception {
        WaterMark waterMark = waterMarkMap.get(timerContext.getCurrentKey());
        for (Map.Entry<String, EventTimer> entry : eventTimerMap.entrySet()) {
            String mapKey = entry.getKey();
            EventTimer mapValue = entry.getValue();
            String currentKey = mapKey.split("_")[1];
            if(waterMark.getTimestamp() > mapValue.getTimestamp()){
                synchronized (timerContext){
                    timerContext.setCurrentKey(currentKey);
                    timerContext.onTimer(waterMark.getTimestamp(), timerContext, timerContext.getProcessContext());
                    deleteEventTimeTimer(mapValue.getTimestamp());
                }
            }
        }
    }

    // 数据 waterMark 并判断是否触发事件
    @Override
    public void waterMark(WaterMark waterMark){
        waterMarkMap.put((String) timerContext.getCurrentKey(), waterMark);
    }
}